package com.uptown.matrix.core.config;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.annotation.Resource;

import com.uptown.matrix.core.consumer.kafka.KafkaConsumerHandler;
import com.uptown.matrix.core.consumer.redis.RedisConsumerHandler;
import com.uptown.matrix.core.consumer.rocketmq.RocketMqConsumerHandler;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;

import com.uptown.matrix.core.annotation.MatrixMessageListener;

@SuppressWarnings("all")
@Slf4j
@Configuration
@Component
public class ConsumerConfigProcessor implements BeanPostProcessor {

    @Resource
    RedisConsumerHandler redisConsumerHandler;

    @Resource
    KafkaConsumerHandler kafkaConsumerHandler;

    @Resource
    RocketMqConsumerHandler rocketMqConsumerHandler;

    @Resource
    Map<String, StreamMessageListenerContainer<String, ObjectRecord<String, Object>>> redisConsumerMap;

    @Resource
    Map<String, KafkaConsumer<String, String>> kafkaConsumerMap;

    @Resource
    Map<String, DefaultMQPushConsumer> rocketMqConsumerMap;

    // 存放kafka topic与consumer的映射
    @Bean(name = "kafkaConsumerMap")
    public Map<String, KafkaConsumer<String, String>> kafkaConsumerMap() {
        return new ConcurrentHashMap<>();
    }

    // 存放redis key与consumer的映射
    @Bean(name = "redisConsumerMap")
    public Map<String, StreamMessageListenerContainer<String, ObjectRecord<String, Object>>> redisConsumerMap() {
        return new ConcurrentHashMap();
    }

    // rocketmq key与consumer的映射
    @Bean(name = "rocketMqConsumerMap")
    public Map<String, DefaultMQPushConsumer> rocketMqConsumerMap() {
        return new ConcurrentHashMap<>();
    }

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @SneakyThrows
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        Method[] methods = ReflectionUtils.getAllDeclaredMethods(bean.getClass());
        for (Method method : methods) {
            MatrixMessageListener annotation = AnnotationUtils.findAnnotation(method, MatrixMessageListener.class);
            if (annotation == null) {
                continue;
            }

            String groupTopicKey = String.format("%s:%s:%s", bean.getClass().getName(), annotation.consumerGroup(), annotation.streamKey());
            String groupTopicCheck = String.format("%s:%s", annotation.consumerGroup(), annotation.streamKey());


            if ("redis".equalsIgnoreCase(annotation.componentType())) {

                if (redisConsumerMap.containsKey(groupTopicKey)) {
                    continue;
                }

                // 校验是否存在同topic、同group的队列 FIXME：分布式
                redisConsumerMap.keySet().forEach(key -> {
                    if (key.contains(groupTopicCheck)) {
                        throw new Error("暂不支持存在多个同topic、group的队列");
                    }
                });

                redisConsumerHandler.createSubscription((StreamListener) bean, annotation, groupTopicKey);
            }
            else if ("kafka".equalsIgnoreCase(annotation.componentType())) {

                if (kafkaConsumerMap.containsKey(groupTopicKey)) {
                    continue;
                }

                // 校验是否存在同topic、同group的队列 FIXME：分布式
                kafkaConsumerMap.keySet().forEach(key -> {
                    if (key.contains(groupTopicCheck)) {
                        throw new Error("暂不支持存在多个同topic、group的队列");
                    }
                });

                kafkaConsumerHandler.createSubscription(bean, annotation, method, groupTopicKey);
            }
            else if ("rocketmq".equalsIgnoreCase(annotation.componentType())) {

                if (rocketMqConsumerMap.containsKey(groupTopicKey)) {
                    continue;
                }

                // 校验是否存在同topic、同group的队列 FIXME：分布式
                rocketMqConsumerMap.keySet().forEach(key -> {
                    if (key.contains(groupTopicCheck)) {
                        throw new Error("暂不支持存在多个同topic、group的队列");
                    }
                });

                rocketMqConsumerHandler.createSubscription(bean, annotation, method, groupTopicKey);
            }
        }
        return bean;
    }
}
